package com.ct.ecommerce.base.core;

import com.ct.ecommerce.api.base.dto.DelayMessageDTO;
import com.google.common.base.Preconditions;
import io.netty.util.Timeout;
import io.netty.util.TimerTask;
import java.util.StringJoiner;
import lombok.extern.slf4j.Slf4j;
import java.util.concurrent.CountDownLatch;
import org.apache.rocketmq.client.producer.SendResult;
import org.apache.rocketmq.common.message.MessageConst;
import org.apache.rocketmq.spring.core.RocketMQTemplate;
import org.springframework.messaging.Message;
import org.springframework.messaging.support.MessageBuilder;

/**
 * @author changtong
 * @date 2021-50-20
 */
@Slf4j
public class SendRealMessageTask implements TimerTask {


    private RocketMQTemplate delayMqProducer;

    private DelayMessageDTO delayMessage;

    private CountDownLatch  countDownLatch;

    private SendResult sendResult;

    public SendRealMessageTask(RocketMQTemplate delayMqProducer,
        DelayMessageDTO delayMessage, CountDownLatch countDownLatch) {
        this.delayMqProducer = delayMqProducer;
        this.delayMessage = delayMessage;
        this.countDownLatch = countDownLatch;
    }

    @Override
    public void run(Timeout timeout) throws Exception {
        validate();
        Message<String> message = buildRealMessage(this.delayMessage);
        String destination = new StringJoiner(":").add(delayMessage.getTopic())
            .add(delayMessage.getTag()).toString();
        try {
            log.info("开始发送延迟消息 destination = {} , message = {}",destination,message);
            sendResult = delayMqProducer.syncSend(destination, message);
        }catch (Exception e){
            log.error("发送延迟消息出错 DelayMessage = " + this.delayMessage + " :",e);
        }
        countDownLatch.countDown();
    }

    private void validate() {
        Preconditions.checkNotNull(delayMqProducer, "producer is  null");
        Preconditions.checkNotNull(delayMessage, "message is  null");
        Preconditions.checkNotNull(countDownLatch, "countDownLatch is null");
    }

    public SendResult getSendResult() {
        return sendResult;
    }

    private Message<String> buildRealMessage(DelayMessageDTO delayMessage) {
        Preconditions.checkNotNull(delayMessage.getTopic(), "message topic is null");
        Preconditions.checkNotNull(delayMessage.getTag(), "message tag is null");
        Preconditions.checkNotNull(delayMessage.getContent(), "message content is null");
        Preconditions.checkNotNull(delayMessage.getKey(), "message content is null");
       return MessageBuilder.withPayload(delayMessage.getContent())
            .setHeader(MessageConst.PROPERTY_KEYS,delayMessage.getKey()).build();
    }


}